package com.wxzhou.wangzhuan.service;

import com.wxzhou.wangzhuan.config.SysConfig;
import com.wxzhou.wangzhuan.constants.CacheNames;
import com.wxzhou.wangzhuan.dto.*;
import com.wxzhou.wangzhuan.dto.reader.ReaderCallBack;
import com.wxzhou.wangzhuan.dto.reader.ReaderRequestInfo;
import com.wxzhou.wangzhuan.dto.reader.ReaderTokenRespDto;
import com.wxzhou.wangzhuan.dto.reader.ReaderUrlRespDto;
import com.wxzhou.wangzhuan.entity.AdBusinessAction;
import com.wxzhou.wangzhuan.entity.AdLog;
import com.wxzhou.wangzhuan.entity.AdsenseAndChannel;
import com.wxzhou.wangzhuan.entity.WzButieRecord;
import com.wxzhou.wangzhuan.event.AdLogEvent;
import com.wxzhou.wangzhuan.event.AdReportLogEvent;
import com.wxzhou.wangzhuan.event.PreAdLogEvent;
import com.wxzhou.wangzhuan.exception.AdReaderException;
import com.wxzhou.wangzhuan.exception.AdReportException;
import com.wxzhou.wangzhuan.repository.AdRespository;
import com.wxzhou.wangzhuan.repository.ButieRespository;
import com.wxzhou.wangzhuan.util.AdvancedGsonPrettyPrint;
import com.wxzhou.wangzhuan.util.ApiSecurityUtils;
import io.quarkus.cache.CacheKey;
import io.quarkus.cache.CacheResult;
import io.smallrye.mutiny.Uni;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.event.Event;
import jakarta.inject.Inject;
import org.jboss.logging.Logger;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;

/**
 * 广告上报服务
 */
@ApplicationScoped
public class AdService {
    private static final Logger LOG = Logger.getLogger(AdService.class);
    @Inject
    AdRespository adRespository;

    @Inject
    AdRateLimitService adRateLimitService;

    @Inject
    ReaderRateLimitService readerRateLimitService;

    @Inject
    AdLogPublishService adLogPublishService;

    @Inject
    SysConfig sysConfig;

    @Inject
    ConfigService configService;

    @Inject
    ReaderApiService readerApiService;

    @Inject
    ReaderRedisService readerRedisService;

    /**
     * 广告日志事件
     */
    @Inject
    Event<PreAdLogEvent> preAdLogEventEvent;

    @Inject
    Event<AdLogEvent> adLogEventEvent;

    @Inject
    Event<AdReportLogEvent> adReportLogEventEvent;

    @Inject
    ButieService butieService; //补贴服务

    @Inject
    ButieRespository butieRespository; //补贴

    /**
     * 批量上报广告收益数据
     * 按每一批的收益，计算本批广告数据的总金额，根据扣除点位、金币换算返回最终结果
     *
     * @param adReportInfos
     * @param currentUserInfo
     * @return
     */
    public Uni<AdReportResponse> reportAd(List<AdReportInfo> adReportInfos, Optional<UserInfoDto> currentUserInfo) {
        if (adReportInfos.size() >= 1000) throw new RuntimeException("上报数据一次性不允许超过1000条");
        //TODO：所有上报数据，必须用户是同一个、并且设备id是同一个，否则视为无效
        //设备ID必须一致，用户id必须一致，否则视为无效
        UserInfoDto userInfoDto = currentUserInfo.get();
        String deviceId = userInfoDto.getDeviceId();
        String appChannel = userInfoDto.getChannel(); //用户携带的appChannel
        Long userId = Long.valueOf(userInfoDto.getUserId());
        //查询所有的sense，并且用于过滤上报数据过滤，通过sense拿到每一个场景对应的AdsenseAndChannel对象，避免手动
        Uni<List<AdsenseAndChannel>> andChannelBySenses = findAdsenseAndChannelBySense(appChannel); //每个sense对应1条数据

        //同一批数据,必须是同一个app_channel,与用户上传的

        //本次上报整个transId（事务发生ID）
        final String transId = PayService.md5(UUID.randomUUID() + String.valueOf(System.currentTimeMillis())).toLowerCase();

        return andChannelBySenses.onItem().ifNull().failWith(() -> new AdReportException("该APP请求不合规！"))
                .onItem().transformToUni(adsenseAndChannels -> {
                    //按sense场景进行分组,
                    final Map<String, AdsenseAndChannel> adsenseAndChannelMap = adsenseAndChannels.stream()
                            .collect(Collectors.groupingBy(
                                    adsenseAndChannel -> String.valueOf(adsenseAndChannel.getSense()),
                                    Collectors.collectingAndThen(Collectors.toList(), list -> list.isEmpty() ? null : list.get(0))));
                    // TODO: 把上报数据按ad_id_sn分组,然后再按sense分组
                    final Map<String, List<AdReportInfo>> groupByAdidsn = adReportInfos.stream().collect(Collectors.groupingBy(AdReportInfo::getAd_id_sn)); //按业务标识分组。

                    //过滤数据并拿到未处理过的上报数据
                    return configService.getSystemConfig().onItem().transformToUni(systemConfigDto -> {
                        return adLogPublishService.findMapByAdidsn(groupByAdidsn.keySet(), userId).onItem().transformToUni(redisAdidsnMap -> {
                            List<AdLog> goodInfos = groupByAdidsn.entrySet().stream().map(adidsnMap -> {
                                        String adidsn = adidsnMap.getKey();
                                        List<AdReportInfo> adInfosFromAdidsn = adidsnMap.getValue();
                                        boolean hasAdidsn = redisAdidsnMap.containsKey(adidsn); //是否存在该广告数据

                                        //如果adidsn在redisAdidsnMap中找不到(hasAdidsn=false),说明整个adInfosFromAdidsn都是可用数据
                                        if (!hasAdidsn) {
                                            // 返回所有该adidsn下的数据，因为这些数据在redis中没有记录，都是未处理的
                                            return adInfosFromAdidsn;
                                        }

                                        //如果adidsn在redisAdidsnMap中找到了(hasAdidsn=true)
                                        //找出所有adInfosFromAdidsn中bs_ac不等于senseListMap.key的所有数据
                                        Set<Integer> bsacList = redisAdidsnMap.get(adidsn);
                                        if (bsacList == null) {
                                            // 如果redis中没有该adidsn的sense数据，返回所有数据
                                            return adInfosFromAdidsn;
                                        }
                                        // 过滤出未处理的数据,找出所有bs_ac不在bsacList中的adInfosFromAdidsn记录
                                        return adInfosFromAdidsn.stream()
                                                .filter(adInfo -> {
                                                    Integer bsAc = Integer.valueOf(adInfo.getBs_ac());
                                                    return !bsacList.contains(bsAc);
                                                })
                                                .collect(Collectors.toList());
                                    }).flatMap(List::stream) // 将所有List<AdReportInfo>扁平化为Stream<AdReportInfo>
                                    .map(info -> {
                                        String senseKey = String.valueOf(info.getSense());
                                        AdsenseAndChannel adsenseAndChannel = adsenseAndChannelMap.get(senseKey);
                                        AdLog adLog = buildBaseAdLog(info,transId, deviceId, userId, adsenseAndChannel);
                                        if (appChannel == null || !appChannel.equalsIgnoreCase(adsenseAndChannel.getApp_channel())) {
                                            throw new AdReportException("app渠道标识不存在或sense不应用于该app");
                                        }
                                        return adLog;
                                    })
                                    .collect(Collectors.toList());

                            Set<Integer> senses = goodInfos.stream().map(adLog -> adLog.getAd_sense()).collect(Collectors.toSet()); //所有场景
//                            if(senses == null || senses.isEmpty()) throw new AdReportException("广告场景不合法");
                            return adRateLimitService.getTodayButieOfUser(userId).onItem()
                                    .transformToUni(userTodayButieMap -> {
                                        return butieService.findAllButieBy(userId, senses)
                                                .onItem().transformToUni(buties -> {
                                                    return butieService.calc(systemConfigDto.getAdMaxAmount(),systemConfigDto.getAdPercent(),goodInfos, buties, userTodayButieMap).onItem().transformToUni(adLogAndButies -> {
                                                        List<AdLog> newGoodInfos = adLogAndButies.stream().map(adLogAndButie -> adLogAndButie.getAdLog()).collect(Collectors.toList()); //日志，已包含补贴
                                                        List<WzButieRecord> wzButieRecords = butieService.calcButieRecord(adLogAndButies, transId);
                                                        String adidsns = newGoodInfos.stream().map(adLog -> adLog.getAd_id_sn()).reduce((s1, s2) -> s1.concat(",").concat(s2)).orElse("");//产生收益的本批编号
                                                        BigDecimal allUserAmount = newGoodInfos.stream().map(adLog -> adLog.getUser_amount()).reduce((d1, d2) -> d1.add(d2)).orElse(BigDecimal.ZERO); //总ecpm，单位分
                                                        BigDecimal amountJinbi = calcUserJinbi(allUserAmount, sysConfig);

                                                        // 根据清洗后的数据，评估本次上报的收益，以及动作，然后异步写入数据库(分2批异步写入，1：异步写入收益数据，发布事件消费。2：异步写入redis，用于入库日志，调用方法publishAdLogsToRedis)
                                                        return executeAsyncOperations(transId, wzButieRecords, adReportInfos, newGoodInfos, allUserAmount, adidsns, appChannel, deviceId, userId).onItem().transform(unused -> AdReportResponse.createNone(amountJinbi))
                                                                .onFailure().recoverWithItem(throwable -> {
                                                                    LOG.error("处理广告上报数据失败", throwable);
                                                                    throw new AdReportException("数据异常");
                                                                });
                                                    });
                                                });
                                    });
                        });
                    });
                }).onFailure().recoverWithUni(throwable -> {
                    LOG.errorf(throwable, "处理广告上报数据失败");
                    throw new AdReportException("执行异常");
                });
    }

    /**
     * 计算用户账变金额，单位分
     * 同时计算了扣除比例
     *
     * @param adMaxAmount
     * @param allEcpmFen(分为ecpm)
     * @param adUserPercent
     * @return
     */
    private BigDecimal calcUserChangeAmountFen(BigDecimal adMaxAmount, BigDecimal allEcpmFen, BigDecimal adUserPercent) {
        //先统一换算成分，再换算成元和金币等单位
        BigDecimal amountFen = allEcpmFen.multiply(adUserPercent).setScale(6, RoundingMode.DOWN); //分.
        //根据日志中的数据,判断用户上报的数据是否满足系统配置的完整收益(广告展示成功,并且拿到ecpm,并且根据ecpm),从上报中找出有ecpm的并且bs_ac=1(展示拿价)的
        BigDecimal finalUserAmountFen = amountFen.compareTo(adMaxAmount) <= 0 ? amountFen : adMaxAmount; //收益控制
        //用户到手的金币=账变金额(分)/100*平台比例（换算成元）
        return finalUserAmountFen;
    }

    /**
     * 计算用户金币数
     *
     * @param finalUserAmountFen
     * @param sysConfig
     * @return
     */
    private BigDecimal calcUserJinbi(BigDecimal finalUserAmountFen, SysConfig sysConfig) {
        Integer convertRatio = sysConfig.getConvertRatio(); //金币换算比例
        BigDecimal amountJinbi = finalUserAmountFen.divide(new BigDecimal("100"), 6, RoundingMode.DOWN).multiply(new BigDecimal(convertRatio)).setScale(0, RoundingMode.DOWN);
        return amountJinbi;
    }


    /**
     * 同步数据库+写入日志
     *
     * @param wzButieRecords
     * @param adReportInfos
     * @param goodInfos
     * @param finalUserAmountFen
     * @param adidsns
     * @param userId
     * @return
     */
    private Uni<Void> executeAsyncOperations(String transId,
                                             List<WzButieRecord> wzButieRecords,
                                             List<AdReportInfo> adReportInfos,
                                             List<AdLog> goodInfos,
                                             BigDecimal finalUserAmountFen,
                                             String adidsns,
                                             String appChannel,
                                             String deviceId,
                                             Long userId) {
        if (adReportInfos.isEmpty() || goodInfos.isEmpty()) return Uni.createFrom().voidItem();
        //先更新缓存
        return adLogPublishService.storeAdInfoBy(adReportInfos, userId)
                .onItem().transformToUni(ignore -> {
                    saveLogAmountAndReturn(transId, finalUserAmountFen, adidsns, userId, AccountChangeType.INCOME.getCode(), "广告收入")
                            .subscribe().with(
                                    unused -> LOG.debugf("保存金额成功"),
                                    throwable -> LOG.errorf(throwable, "保存金额失败"));
                    //保存补贴
                    saveAdButieRecords(wzButieRecords,transId)
                            .subscribe().with(
                                    unused -> LOG.debugf("保存补贴记录完成"),
                                    throwable -> LOG.errorf(throwable, "保存补贴记录异常"));

                    //更新用户今日累计补贴上限
                    adRateLimitService.storeTodayButieOfUser(userId, wzButieRecords)
                            .subscribe().with(
                                    unused -> LOG.debugf("保存金额成功"),
                                    throwable -> LOG.errorf(throwable, "保存金额失败"));

                    //写入日志.
//                    Uni<Void> publishTask = publishAdLog(goodInfos)
//                            .onFailure().invoke(e -> LOG.error("发布日志失败", e));
                    adLogEventEvent.fireAsync(new AdLogEvent(goodInfos));
                    adReportLogEventEvent.fireAsync(new AdReportLogEvent(adReportInfos, appChannel, deviceId, userId));
                    return Uni.createFrom().voidItem();
                })
                .onFailure().invoke(e -> LOG.error("存储广告信息失败", e));
    }

    private Uni<Void> saveAdButieRecords(List<WzButieRecord> wzButieRecords,String transId) {
        return butieRespository.batchInsertButieRecord(wzButieRecords,transId)
                .onItem().invoke(unused -> LOG.infof("保存补贴记录完成!"))
                .onFailure().invoke(throwable -> LOG.errorf("保存补贴记录异常: %s", throwable.getMessage()))
                .replaceWithVoid();
    }

    /**
     * 异步存储用户收益，将收益归集到主账号、并同步账变。
     *
     * @param userid
     * @return
     */
    private Uni<Void> saveLogAmountAndReturn(String transId, BigDecimal amountFen, String adidsns, Long userid, int changeType, String remark) {
        return adRespository.saveLogAmountAndReturn(transId, amountFen, adidsns, userid, changeType, remark)
                .onItem().invoke(unused -> LOG.infof("收益同步完成!"))
                .onFailure().invoke(throwable -> LOG.errorf("收益同步异常: %s", throwable.getMessage()))
                .replaceWithVoid();
    }

    /**
     * 推送广告日志消息到redis订阅，通过redis消费者进行广告日志数据异步批量入库。
     * 优化空间：高并发情况下，实时数据入库是非常消耗性能，会导致系统负载加大，应该直接将日志数据（不太重要）推送到缓存中。并通过定时任务，批量写入到stream中，只消费1次并删除数据，streams的key是adlog:events（只记录）
     * 如果服务重启，则自动从
     *
     * @param adLogList
     * @return
     */
//    private Uni<Void> publishAdLog(List<AdLog> adLogList) {
//        if (adLogList == null || adLogList.size() <= 0) {
//            return Uni.createFrom().voidItem();
//        }
//        return adLogService.sendAdlogs(adLogList)
//                .onItem().invoke(messageId ->
//                        LOG.debugf("成功发布数据到Redis Stream，消息ID"))
//                .onFailure().invoke(throwable ->
//                        LOG.errorf(throwable, "发布数据到Redis Stream失败"))
//                .replaceWithVoid();
//    }

    /**
     * 构建基础AdLog对象
     */
    private AdLog buildBaseAdLog(AdReportInfo reportInfo, String transId, String deviceId, Long userId, AdsenseAndChannel adsenseAndChannel) {
        int sense = adsenseAndChannel.getSense();
        String adid = adsenseAndChannel.getAdid();
        int adChannel = adsenseAndChannel.getAd_channel();
        Long adsenseid = adsenseAndChannel.getAdsenseid();
        String app_channel = adsenseAndChannel.getApp_channel();
        int adType = adsenseAndChannel.getAd_type();
        int ad_nt = adsenseAndChannel.getAd_nt();
        Integer bsAc = reportInfo.getBs_ac(); //发生时机
        AdLog.Builder adlogBuilder = AdLog.buildReport(adid, reportInfo.getAd_id_sn(), userId, deviceId, adChannel, ad_nt, sense, bsAc)
                .adType(adType)
                .appChannel(app_channel)
                .reportTime(reportInfo.getReportTime())
                .userAmount(new BigDecimal(reportInfo.getEcpm()).divide(new BigDecimal("1000"), 6, RoundingMode.DOWN))
                .adIdPk(adsenseid)
                .transId(transId)
                .adInfo(reportInfo.getAdinfo() != null ? reportInfo.getAdinfo() : "")// 设置广告源数据
                .remark("ad上报");
        //广告时机=成功展示才获取并设置ecpm,其他情况下没有价格
        if (bsAc == AdBusinessAction.SUCCESS_SHOW.getCode()) {
            adlogBuilder.ecpm(reportInfo.getEcpm());
        }
        AdLog adLog = adlogBuilder.build();
        return adLog;
    }

    //根据sense查找广告媒体、渠道包标识、广告场景、广告类型、广告渠道
    //todo：线上需要增加缓存，同一个场景拉取的广告等配置是一致的,后台凡是设计到修改sense时，都需要清空该值
    @CacheResult(cacheName = CacheNames.CACHE_SENSE_ADINFO_CONFIG)
    public Uni<AdsenseAndChannel> findAdsenseAndChannelBySense(@CacheKey int sense) {
        return adRespository.findAdsenseAndChannelBySense(sense).onItem().ifNull().failWith(() -> new AdReportException("场景不存在 not found : " + sense));
    }

    //TODO：获取app的标识渠道，注意后台变更要同步修改
    @CacheResult(cacheName = CacheNames.CACHE_CHANNEL_ADINFO_CONFIG)
    public Uni<List<AdsenseAndChannel>> findAdsenseAndChannelBySense(@CacheKey String appchannel) {
        return adRespository.findAllAdsenseByChannel(appchannel).onItem().ifNull().failWith(() -> new AdReportException("appchannel not found : " + appchannel));
    }

    /**
     * 广告预拉取，步骤如下：
     * 1：记录到wz_ad_log中，用于记录广告拉取链路，并返回生成的ad_id_sn（行为分组id，根据广告位+动作）
     *
     * @param adPreReportInfo
     * @param currentUserInfo
     * @return 本次预拉取响应对象
     */
    public Uni<AdPreReportResponse> preReportAd(AdPreReportInfo adPreReportInfo, Optional<UserInfoDto> currentUserInfo) {
        LOG.debugf("用户请求:%s", AdvancedGsonPrettyPrint.prettyPrint(currentUserInfo.get()));
        int sense = adPreReportInfo.getSense();
        UserInfoDto userInfoDto = currentUserInfo.get();
        String appChannel = userInfoDto.getChannel(); //app渠道标识
        String deviceId = userInfoDto.getDeviceId(); //设备id

        //根据sense查找广告媒体、渠道包标识、广告场景、广告类型、广告渠道,然后校验合法性,如果不合法,直接抛出异常,比如:渠道包标识不匹配等!校验完成后,异步执行最后异步同步数据库
        return findAdsenseAndChannelBySense(sense).onItem().transformToUni(adsenseAndChannel -> {
            String app_channel = adsenseAndChannel.getApp_channel();
            //校验用户请求的header的app渠道和获取到的sense对应的渠道是否匹配!
            if (appChannel == null || !appChannel.equalsIgnoreCase(app_channel)) {
                throw new AdReportException("app渠道标识不存在或sense不应用于该app");
            }
            return Uni.createFrom().item(adsenseAndChannel);
        }).onItem().transformToUni(adsenseAndChannel -> {
            // 检查是否允许拉取广告、拉取频率限制等，如果不满足，直接返回false
            return genericAdIdSnFrom(adsenseAndChannel, System.currentTimeMillis(), appChannel, deviceId)
                    .onItem()
                    .transformToUni(adIdSn -> {
                        return adRateLimitService.isAllowed(adIdSn, sense, adsenseAndChannel.getApp_channel(), deviceId, 0, userInfoDto)
                                .onItem().transform(isAllow -> {
                                    if (!isAllow) {
                                        throw new AdReportException("超过广告拉取频率,请稍后拉取");
                                    }
                                    //异步入库
                                    preAdLogEventEvent.fireAsync(new PreAdLogEvent(userInfoDto, adsenseAndChannel, adIdSn, "AD预拉"));

                                    AdReportInfo adReportInfo = new AdReportInfo();
//                                    adReportInfo.setAd_id_sn("a36d6540f458ad4f3135b244df3f661e");
                                    adReportInfo.setAd_id_sn(adIdSn);
                                    adReportInfo.setSense(sense);
//                                    adReportInfo.setBs_ac(AdBusinessAction.SUCCESS_SHOW.getCode());
                                    adReportInfo.setBs_ac(AdBusinessAction.PRE.getCode());
                                    List<AdReportInfo> adReportInfos = Arrays.asList(adReportInfo);
                                    adLogPublishService.storeAdInfoBy(adReportInfos, userInfoDto.getUserId())
                                            .subscribe().with(
                                                    isSuccess -> LOG.debugf("Redis存储成功: {}", adIdSn),
                                                    isFailure -> LOG.errorf("Redis存储失败: {}", adIdSn, isFailure)
                                            );
                                    AdPreReportResponse response = new AdPreReportResponse(adIdSn, adsenseAndChannel.getAdid(), true);
                                    return response;
                                });
                    });

        });
    }

    /**
     * 生成预请求的场景ad_id_sn
     *
     * @param adsenseAndChannelObj 广告位 不允许为空
     * @param channel              广告渠道 不允许为空
     * @param deviceid             设备ID 不允许为空
     * @return
     */
    private Uni<String> genericAdIdSnFrom(AdsenseAndChannel adsenseAndChannelObj, long currentTimeMillis, String channel, String deviceid) {
// 验证参数是否为空
        if (channel == null || channel.trim().isEmpty()) {
            return Uni.createFrom().failure(new AdReportException("Channel cannot be null or empty"));
        }
        if (deviceid == null || deviceid.trim().isEmpty()) {
            return Uni.createFrom().failure(new AdReportException("Device ID cannot be null or empty"));
        }

        return Uni.createFrom().item(adsenseAndChannelObj)
                .onItem().transform(adsenseandchannel -> {
                    // 验证adsenseId是否为空
                    if (adsenseandchannel == null || adsenseandchannel.getAdid().trim().isEmpty()) {
                        throw new AdReportException("Adsense ID cannot be null or empty");
                    }

                    // 生成随机偏移值（6位随机数）
                    int randomOffset = ThreadLocalRandom.current().nextInt(100000, 999999);

                    // 拼接唯一明文数据
                    String data = String.format("%s|%s|%s|%d|%d", adsenseandchannel.getSense(), channel.trim(), deviceid.trim(), currentTimeMillis, randomOffset);

                    // 生成MD5哈希值
                    String hash = ApiSecurityUtils.md5Hash(data);
                    if (hash == null || hash.isEmpty()) {
                        throw new RuntimeException("Failed to generate MD5 hash");
                    }
                    return hash.toLowerCase();
                });
    }

    /**
     * 批量插入广告数据到数据库
     *
     * @param adLogs
     * @return
     */
    public Uni<Integer> batchInsert(List<AdLog> adLogs) {
        return adRespository.batchInsert(adLogs);
    }

    private Uni<String> refreshTokenAndSave(String app_channel, String deviceId, String adIdSn, int sense, Long userId) {
        return readerApiService.getToken(userId).onItem().transformToUni(dto -> {
            if (dto == null || !dto.isSuccess()) {
                throw new AdReaderException("获取阅读令牌失败");
            }
            ReaderTokenRespDto respDto = dto.getData();
            //成功获取到token,把token和用户请求关联写入到redis
            return readerRedisService.storeReaderToken(app_channel, deviceId, adIdSn, sense, respDto.getToken(), userId).onItem().transform(aBoolean -> {
                return respDto.getToken();
            });

        });
    }


    /**
     * 获取最新阅读地址
     *
     * @param userId
     * @param preJkey
     * @param token
     * @return
     */
    private Uni<ReaderUrlRespDto> getUrlBy(String app_channel, String deviceId, String adIdSn, int sense, Long userId, String preJkey, String token) {
        if (token == null) throw new AdReaderException("获取阅读令牌失败");
        return readerApiService.getUrl(userId, preJkey, token).onItem()
                .transform(readerUrlResp -> {
                    boolean success = readerUrlResp.isSuccess();
                    if (success == true && readerUrlResp.getData() != null && readerUrlResp.getData().getUrl() != null) {
                        ReaderUrlRespDto data = readerUrlResp.getData();
                        String thisJkey = data.getJkey(); //把jkey存到最新的token里面
                        //TODO:发送事件把阅读的返回结果存储到redis中
                        readerRedisService.storeReaderJkey(app_channel, deviceId, adIdSn, sense, token, thisJkey, userId).subscribe().with(aBoolean -> {
                            LOG.debugf("更新缓存成功");
                        }, throwable -> LOG.errorf(throwable, "更新缓存失败"));
                        return data;
                    }
                    throw new AdReaderException("获取阅读地址失败");
                });

    }

    /**
     * 根据预拉的adidsn和sense,去拿跳转链接.
     *
     * @param readerRequestInfo
     * @param currentUserInfo
     * @return
     */
    public Uni<AdPreReportResponse> getReaderInfo(ReaderRequestInfo readerRequestInfo, Optional<UserInfoDto> currentUserInfo) {
        LOG.debugf("用户%s阅读请求:%s", currentUserInfo.get().getUserId(), AdvancedGsonPrettyPrint.prettyPrint(readerRequestInfo));
        int sense = readerRequestInfo.getSense();
        UserInfoDto userInfoDto = currentUserInfo.get();
        String appChannel = userInfoDto.getChannel(); //app渠道标识
        String deviceId = userInfoDto.getDeviceId(); //设备id
        Long userId = userInfoDto.getUserId();

        //根据sense查找广告媒体、渠道包标识、广告场景、广告类型、广告渠道,然后校验合法性,如果不合法,直接抛出异常,比如:渠道包标识不匹配等!校验完成后,异步执行最后异步同步数据库
        return findAdsenseAndChannelBySense(sense).onItem().transformToUni(adsenseAndChannel -> {
            String app_channel = adsenseAndChannel.getApp_channel();
            //校验用户请求的header的app渠道和获取到的sense对应的渠道是否匹配!
            if (appChannel == null || !appChannel.equalsIgnoreCase(app_channel)) {
                throw new AdReaderException("app渠道标识不存在或sense不应用于该app");
            }
            return Uni.createFrom().item(adsenseAndChannel);
        }).onItem().transformToUni(adsenseAndChannel -> {
            // 检查是否允许拉取广告、拉取频率限制等，如果不满足，直接返回false
            return genericAdIdSnFrom(adsenseAndChannel, System.currentTimeMillis(), appChannel, deviceId)
                    .onItem()
                    .transformToUni(adIdSn -> {
                        return readerRateLimitService.isAllowed(adIdSn, sense, adsenseAndChannel.getApp_channel(), deviceId, 0, userInfoDto)
                                .onItem().transformToUni(isAllow -> {
                                    if (!isAllow) {
                                        throw new AdReaderException("超过广告拉取频率,请稍后拉取");
                                    }
                                    //异步通知,预拉数据入库!
                                    preAdLogEventEvent.fireAsync(new PreAdLogEvent(userInfoDto, adsenseAndChannel, adIdSn, "阅读预拉"));

                                    return readerRedisService.findTokenAllBy(appChannel, deviceId, readerRequestInfo.getSense(), userId)
                                            .onItem().transformToUni(token -> {
                                                if (token == null || token.equals("")) {
                                                    return refreshTokenAndSave(appChannel, deviceId, adIdSn, sense, userId).onItem().transformToUni(newtoken -> {
                                                        if (newtoken != null) {
                                                            return getUrlBy(appChannel, deviceId, adIdSn, sense, userId, null, newtoken).onItem().transform(readerUrlRespDto -> {
                                                                return new AdPreReportResponse(adIdSn, readerUrlRespDto.getUrl(), true);
                                                            });
                                                        }
                                                        return Uni.createFrom().item(new AdPreReportResponse());//拉取失败
                                                    });
                                                }
                                                return readerRedisService.findJkeyAllBy(token, userId).onItem().transformToUni(jkey -> {
                                                    return getUrlBy(appChannel, deviceId, adIdSn, sense, userId, jkey, token).onItem().transform(readerUrlRespDto -> {
                                                        return new AdPreReportResponse(adIdSn, readerUrlRespDto.getUrl(), true);
                                                    });
                                                });
                                            });
                                });
                    });

        });
    }

    /**
     * 回调
     *
     * @param readerCallBack
     * @return
     */
    public Uni<Object> readerCallBack(ReaderCallBack readerCallBack) {
        String uid = readerCallBack.getUid();
        String jkey = readerCallBack.getJkey();
        final String transId = PayService.md5(UUID.randomUUID() + String.valueOf(System.currentTimeMillis())).toLowerCase();
        return readerRedisService.findAndDelBy(uid, jkey)
                .onItem().transformToUni(result -> {
                    if (result == null || result.split("_").length < 4) throw new AdReaderException("数据已过期");
                    String[] s = result.split("_");
                    if (s.length == 5) {
                        String deviceId = s[0];
                        String app_channel = s[1];
                        int sense = Integer.valueOf(s[2]);
                        String adidsn = s[3];
                        return findAdsenseAndChannelBySense(sense).onItem().transformToUni(adsenseAndChannel -> {
                            //写入到redis,方便后续拦截判断.
                            int bsAc = AdBusinessAction.SUCCESS.getCode();
                            return readerRateLimitService.recordAdRequest(adidsn, sense, app_channel, deviceId, bsAc, uid)
                                    .onItem().transformToUni(unused -> {

                                        return configService.getSystemConfig().onItem().transform(systemConfigDto -> {
                                            BigDecimal adMaxAmount = systemConfigDto.getAdMaxAmount();//最高价，单位分
                                            BigDecimal adUserPercent = systemConfigDto.getAdPercent();
                                            BigDecimal allEcpmFen = new BigDecimal(systemConfigDto.getReaderEcpm()).divide(new BigDecimal("1000"), 6, RoundingMode.DOWN); //TODO：阅读收益是固定价格，按千次给，比如50，
                                            BigDecimal userChangeAmountFen = calcUserChangeAmountFen(adMaxAmount, allEcpmFen, adUserPercent);

                                            //TODO：实时账变入库,5标识
                                            saveLogAmountAndReturn(transId, userChangeAmountFen, adidsn, Long.valueOf(uid), 5, "阅读收入")
                                                    .subscribe()
                                                    .with(
                                                            unused1 -> LOG.debugf(""),
                                                            throwable -> LOG.errorf(throwable, "")
                                                    );

                                            //发送通知,写入日志
                                            AdLog adLog = buildReaderAdLog(adidsn, bsAc, systemConfigDto.getReaderEcpm(), deviceId, Long.valueOf(uid), adsenseAndChannel);
                                            adLogEventEvent.fireAsync(new AdLogEvent(Arrays.asList(adLog)));
                                            return "ok";
                                        });

                                    });
                        });
                    }
                    //入库失败
                    return Uni.createFrom().item("error");
                });
    }

    /**
     * 构造阅读入库对象
     *
     * @param adidsn
     * @param bsAc
     * @param ecpm
     * @param deviceId
     * @param userId
     * @param adsenseAndChannel
     * @return
     */
    private AdLog buildReaderAdLog(String adidsn, int bsAc, Integer ecpm, String deviceId, Long userId, AdsenseAndChannel adsenseAndChannel) {
        int sense = adsenseAndChannel.getSense();
        String adid = adsenseAndChannel.getAdid();
        int adChannel = adsenseAndChannel.getAd_channel();
        Long adsenseid = adsenseAndChannel.getAdsenseid();
        String app_channel = adsenseAndChannel.getApp_channel();
        int adType = adsenseAndChannel.getAd_type();
        int ad_nt = adsenseAndChannel.getAd_nt();

        AdLog adLog = AdLog.buildReport(adid, adidsn, userId, deviceId, adChannel, ad_nt, sense, bsAc)
                .adType(adType)
                .appChannel(app_channel)
                .adIdPk(adsenseid)
                .ecpm(ecpm)
                .adInfo("")// 设置广告源数据
                .remark("阅读")
                .build();
        return adLog;
    }

}
